package site.teamo.biu.flink.sdk.job;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ListUtil;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author haocongshun
 * @date 2023/07/26 11:16:00
 */
public class JobUtil {

    public static List<Pipeline> toPipelines(JobInfo info) {
        List<JobNodeInfo> sourceNodeInfos = getSourceNodeInfos(info.getNodes());
        return sourceNodeInfos.stream()
                .map(n -> new Pipeline(getChildrenNodeInfos(info.getNodes(), n), n))
                .collect(Collectors.toList());
    }

    private static List<JobNodeInfo> getSourceNodeInfos(List<JobNodeInfo> jobNodeInfos) {
        Set<String> referencedJobNodeNames = jobNodeInfos.stream()
                .flatMap(n -> Optional.ofNullable(n.getNextNode()).orElse(new ArrayList<>()).stream())
                .collect(Collectors.toSet());
        return jobNodeInfos.stream()
                .filter(n -> !referencedJobNodeNames.contains(n.getName()))
                .collect(Collectors.toList());
    }

    private static List<JobNodeInfo> getChildrenNodeInfos(List<JobNodeInfo> jobNodeInfos, JobNodeInfo node) {
        if (CollectionUtil.isEmpty(node.getNextNode())) {
            return ListUtil.of(node);
        }
        List<JobNodeInfo> list = jobNodeInfos.stream()
                .filter(n -> node.getNextNode().contains(n.getName()))
                .flatMap(n -> getChildrenNodeInfos(jobNodeInfos, n).stream())
                .collect(Collectors.toList());
        list.add(node);
        return list;
    }

}
